package com.tom.architect02.mnbase.annotation.jvms.juc.juc4;

import java.util.HashSet;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

// 自定义线程池。
public class ClassThreadPool {

    //任务队列
    private ClassBlockingQueue<Runnable> taskQueue;
    //线程容器
    private HashSet<Worker> workers;
    //核心线程数
    private AtomicInteger coreSize;

    //超时时间
    private long timeout;
    //时间单位
    private TimeUnit unit;
    private RejectPolicy rejectPolicy;

    //size + state， 合并到一起。前三位是状态数据，后29位是size。
    //111 0000000000000000000000000011
    private AtomicInteger ctl;

    public ClassThreadPool(int coreSize, int taskSize, long timeout, TimeUnit unit, RejectPolicy rejectPolicy) {
        this.coreSize = new AtomicInteger(coreSize);
        this.timeout = timeout;
        this.unit = unit;
        this.taskQueue = new ClassBlockingQueue<Runnable>(new AtomicInteger(taskSize));
        this.workers = new HashSet<>();
//        this.rejectPolicy = new WaitRejectPolicy(); // 默认策略
        this.rejectPolicy = rejectPolicy;
    }

    public void executor(Runnable task) {
        if (coreSize.intValue() > workers.size()) {
            Worker worker = new Worker(task);
            worker.start();
            workers.add(worker);
        } else {
//            taskQueue.put(task);//我能不能拒绝掉过来的任务？
            taskQueue.tryPut(rejectPolicy, task); // rejectPolicy拒绝策略
            //策略模式
            //拒绝的场景
            //1.死等
            //2.带超时等待
            //3.放弃任务执行
            //4.让调用者抛出异常
            //5.让调用者自己去执行
        }
    }

    class Worker extends Thread {
        private Runnable task;
        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
//            while(task != null || (task = taskQueue.take()) != null) {
            while (task != null || (task = taskQueue.poll(timeout, unit)) != null) {
                try {
                    task.run();
                } finally {
                    task = null; // 重置
                }
            }
        }
    }

    public static void main(String[] args) {
        ClassThreadPool threadPool = new ClassThreadPool(2, 10, 1000, TimeUnit.MILLISECONDS,
                // 策略
                (queue, task) -> {
                    //死等方案
//                    queue.put(task);
                    //2.带超时等待
                    queue.poll(1000, TimeUnit.MILLISECONDS);
                    //3.放弃任务执行
                    System.out.println("放弃执行..");
                    //4.让调用者抛出异常
                    new RuntimeException("队列已满....");
                    //5.让调用者自己去执行
                }
        );
        for (int i = 0; i < 10; i++) {
            int j = i;
            threadPool.executor(() -> {
                System.out.println("Thread:" + Thread.currentThread() + "value:" + j);
            });
        }

    }

}
